package com.datareport.controller;

import com.datareport.common.response.Result;
import com.datareport.compute.FlinkJobManager;
import com.datareport.service.FlinkComputeService;
import com.datareport.domain.entity.StatRule;
import com.datareport.repository.StatRuleRepository;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Flink计算控制器
 * 
 * 提供Flink作业管理的REST API接口，包括作业生命周期管理、状态监控、规则解析等功能。
 * 所有接口统一返回Result对象，包含操作结果和错误信息。
 * 
 * @author liyc
 * @since 2024-01-01
 */
@Slf4j
@RestController
@RequestMapping("/api/flink")
@Api(tags = "Flink计算管理")
public class FlinkComputeController {

    @Autowired
    private FlinkJobManager jobManager;

    @Autowired
    private FlinkComputeService computeService;

    @Autowired
    private StatRuleRepository ruleRepository;

    /**
     * 启动规则计算作业
     * 
     * 根据规则ID启动对应的Flink计算作业，作业启动后会进行实时数据统计分析
     * 
     * @param ruleId 规则ID，对应StatRule表中的主键
     * @return Result对象，包含作业ID和成功信息，失败时返回错误信息
     */
    @PostMapping("/job/start/{ruleId}")
    @ApiOperation("启动规则计算作业")
    public Result<String> startJob(
            @ApiParam(value = "规则ID", required = true, example = "rule_001") 
            @PathVariable String ruleId) {
        try {
            log.info("开始启动规则计算作业: ruleId={}", ruleId);
            String jobId = jobManager.startJob(ruleId);
            log.info("规则计算作业启动成功: ruleId={}, jobId={}", ruleId, jobId);
            return Result.success(jobId, "作业启动成功");
        } catch (Exception e) {
            log.error("启动规则计算作业失败: ruleId={}", ruleId, e);
            return Result.fail("作业启动失败: " + e.getMessage());
        }
    }

    /**
     * 停止规则计算作业
     *
     * 根据规则ID停止正在运行的Flink计算作业
     *
     * @param ruleId 规则ID
     * @return Result对象，成功时返回null数据和成功信息，失败时返回错误信息
     */
    @PostMapping("/job/stop/{ruleId}")
    @ApiOperation("停止规则计算作业")
    public Result<String> stopJob(
            @ApiParam(value = "规则ID", required = true, example = "rule_001")
            @PathVariable String ruleId) {
        try {
            log.info("开始停止规则计算作业: ruleId={}", ruleId);
            jobManager.stopJob(ruleId);
            log.info("规则计算作业停止成功: ruleId={}", ruleId);
            return Result.success(null, "作业停止成功");
        } catch (Exception e) {
            log.error("停止规则计算作业失败: ruleId={}", ruleId, e);
            return Result.fail("作业停止失败: " + e.getMessage());
        }
    }

    /**
     * 重启规则计算作业
     *
     * 先停止再启动规则计算作业，用于配置更新后的重新计算
     *
     * @param ruleId 规则ID
     * @return Result对象，包含新的作业ID和成功信息，失败时返回错误信息
     */
    @PostMapping("/job/restart/{ruleId}")
    @ApiOperation("重启规则计算作业")
    public Result<String> restartJob(
            @ApiParam(value = "规则ID", required = true, example = "rule_001")
            @PathVariable String ruleId) {
        try {
            log.info("开始重启规则计算作业: ruleId={}", ruleId);
            String jobId = jobManager.restartJob(ruleId);
            log.info("规则计算作业重启成功: ruleId={}, jobId={}", ruleId, jobId);
            return Result.success(jobId, "作业重启成功");
        } catch (Exception e) {
            log.error("重启规则计算作业失败: ruleId={}", ruleId, e);
            return Result.fail("作业重启失败: " + e.getMessage());
        }
    }

    /**
     * 获取单个作业状态
     *
     * 查询指定规则ID对应的Flink作业当前运行状态
     *
     * @param ruleId 规则ID
     * @return Result对象，包含作业状态信息，失败时返回错误信息
     */
    @GetMapping("/job/status/{ruleId}")
    @ApiOperation("获取作业状态")
    public Result<FlinkJobManager.JobStatusInfo> getJobStatus(
            @ApiParam(value = "规则ID", required = true, example = "rule_001")
            @PathVariable String ruleId) {
        try {
            FlinkJobManager.JobStatusInfo status = jobManager.getJobStatus(ruleId);
            log.debug("获取作业状态成功: ruleId={}, status={}", ruleId, status);
            return Result.success(status);
        } catch (Exception e) {
            log.error("获取作业状态失败: ruleId={}", ruleId, e);
            return Result.fail("获取作业状态失败: " + e.getMessage());
        }
    }

    /**
     * 获取所有作业状态
     *
     * 查询系统中所有Flink作业的当前运行状态列表
     *
     * @return Result对象，包含所有作业状态列表，失败时返回错误信息
     */
    @GetMapping("/job/status")
    @ApiOperation("获取所有作业状态")
    public Result<List<FlinkJobManager.JobStatusInfo>> getAllJobStatuses() {
        try {
            Map<String, FlinkJobManager.JobStatusInfo> statuses = jobManager.getAllJobStatuses();
            List<FlinkJobManager.JobStatusInfo> statusList = statuses.values().stream()
                    .collect(Collectors.toList());
            log.debug("获取所有作业状态成功，共{}个作业", statusList.size());
            return Result.success(statusList);
        } catch (Exception e) {
            log.error("获取所有作业状态失败", e);
            return Result.fail("获取所有作业状态失败: " + e.getMessage());
        }
    }

    /**
     * 获取Flink集群信息
     *
     * 查询Flink集群的运行状态和配置信息
     *
     * @return Result对象，包含集群信息，失败时返回错误信息
     */
    @GetMapping("/cluster/info")
    @ApiOperation("获取Flink集群信息")
    public Result<FlinkJobManager.ClusterInfo> getClusterInfo() {
        try {
            FlinkJobManager.ClusterInfo clusterInfo = jobManager.getClusterInfo();
            log.debug("获取集群信息成功: {}", clusterInfo);
            return Result.success(clusterInfo);
        } catch (Exception e) {
            log.error("获取集群信息失败", e);
            return Result.fail("获取集群信息失败: " + e.getMessage());
        }
    }

    /**
     * 获取可启用的规则列表
     *
     * 查询数据库中状态为启用（status=1）的所有统计规则
     *
     * @return Result对象，包含可用规则列表，失败时返回错误信息
     */
    @GetMapping("/rules/available")
    @ApiOperation("获取可启用的规则列表")
    public Result<List<StatRule>> getAvailableRules() {
        try {
            List<StatRule> rules = ruleRepository.selectList(null);
            List<StatRule> availableRules = rules.stream()
                    .filter(rule -> rule.getStatus() == 1)
                    .collect(Collectors.toList());
            log.debug("获取可用规则列表成功，共{}个可用规则", availableRules.size());
            return Result.success(availableRules);
        } catch (Exception e) {
            log.error("获取可用规则列表失败", e);
            return Result.fail("获取可用规则列表失败: " + e.getMessage());
        }
    }

    /**
     * 批量启动规则计算作业
     *
     * 根据提供的规则ID列表批量启动对应的Flink计算作业
     * 使用并行流提高批量处理效率，每个规则独立处理，互不影响
     *
     * @param ruleIds 规则ID列表
     * @return Result对象，包含每个规则的启动结果Map，失败时返回错误信息
     */
    @PostMapping("/job/batch-start")
    @ApiOperation("批量启动规则计算作业")
    public Result<Map<String, String>> batchStartJobs(
            @ApiParam(value = "规则ID列表", required = true, example = "[\"rule_001\", \"rule_002\"]")
            @RequestBody List<String> ruleIds) {
        try {
            log.info("开始批量启动规则计算作业，共{}个规则", ruleIds.size());
            Map<String, String> results = ruleIds.parallelStream()
                    .collect(Collectors.toMap(
                            ruleId -> ruleId,
                            ruleId -> {
                                try {
                                    String jobId = jobManager.startJob(ruleId);
                                    log.info("批量启动作业成功: ruleId={}, jobId={}", ruleId, jobId);
                                    return jobId;
                                } catch (Exception e) {
                                    log.error("批量启动作业失败: ruleId={}", ruleId, e);
                                    return "FAILED: " + e.getMessage();
                                }
                            }
                    ));
            log.info("批量启动规则计算作业完成");
            return Result.success("批量启动作业完成", results);
        } catch (Exception e) {
            log.error("批量启动作业失败", e);
            return Result.fail("批量启动作业失败: " + e.getMessage());
        }
    }

    /**
     * 批量停止规则计算作业
     *
     * 根据提供的规则ID列表批量停止正在运行的Flink计算作业
     * 使用并行流提高批量处理效率，每个规则独立处理，互不影响
     *
     * @param ruleIds 规则ID列表
     * @return Result对象，包含每个规则的停止结果Map，失败时返回错误信息
     */
    @PostMapping("/job/batch-stop")
    @ApiOperation("批量停止规则计算作业")
    public Result<Map<String, String>> batchStopJobs(
            @ApiParam(value = "规则ID列表", required = true, example = "[\"rule_001\", \"rule_002\"]")
            @RequestBody List<String> ruleIds) {
        try {
            log.info("开始批量停止规则计算作业，共{}个规则", ruleIds.size());
            Map<String, String> results = ruleIds.parallelStream()
                    .collect(Collectors.toMap(
                            ruleId -> ruleId,
                            ruleId -> {
                                try {
                                    jobManager.stopJob(ruleId);
                                    log.info("批量停止作业成功: ruleId={}", ruleId);
                                    return "SUCCESS";
                                } catch (Exception e) {
                                    log.error("批量停止作业失败: ruleId={}", ruleId, e);
                                    return "FAILED: " + e.getMessage();
                                }
                            }
                    ));
            log.info("批量停止规则计算作业完成");
            return Result.success("批量停止作业完成", results);
        } catch (Exception e) {
            log.error("批量停止作业失败", e);
            return Result.fail("批量停止作业失败: " + e.getMessage());
        }
    }

    /**
     * 清理作业状态
     *
     * 清除指定规则ID的作业状态缓存信息
     *
     * @param ruleId 规则ID
     * @return Result对象，成功时返回null数据和成功信息，失败时返回错误信息
     */
    @DeleteMapping("/job/status/{ruleId}")
    @ApiOperation("清理作业状态")
    public Result<Void> clearJobStatus(
            @ApiParam(value = "规则ID", required = true, example = "rule_001")
            @PathVariable String ruleId) {
        try {
            log.info("开始清理作业状态: ruleId={}", ruleId);
            jobManager.clearJobStatus(ruleId);
            log.info("作业状态清理成功: ruleId={}", ruleId);
            return Result.success("作业状态清理成功", null);
        } catch (Exception e) {
            log.error("清理作业状态失败: ruleId={}", ruleId, e);
            return Result.fail("清理作业状态失败: " + e.getMessage());
        }
    }

    /**
     * 测试规则解析
     *
     * 测试规则配置的解析功能，验证规则配置的正确性
     * 注意：该方法调用的parseRule在FlinkComputeService中不存在，
     * 实际应该调用parseRuleConfig方法或者使用RuleParser类
     *
     * @param ruleConfig 规则配置JSON字符串
     * @return Result对象，包含解析结果，失败时返回错误信息
     */
    @PostMapping("/rule/parse")
    @ApiOperation("测试规则解析")
    public Result<Object> testRuleParse(
            @ApiParam(value = "规则配置", required = true,
                     example = "{\"tables\":\"[]\",\"conditions\":\"[]\",\"groupFields\":\"[]\",\"aggFields\":\"[]\"}")
            @RequestBody String ruleConfig) {
        try {
            log.info("开始测试规则解析");
            // TODO: 修复方法调用 - parseRule方法不存在，应该使用parseRuleConfig
            // Object result = computeService.parseRule(ruleConfig);

            // 临时解决方案：返回规则配置字符串
            log.warn("parseRule方法不存在，返回原始配置。建议使用RuleParser.parseToFlinkSql进行规则解析");
            return Result.success(ruleConfig, "规则解析成功（注意：当前为测试实现）");
        } catch (Exception e) {
            log.error("规则解析失败", e);
            return Result.fail("规则解析失败: " + e.getMessage());
        }
    }
}